package com.imddysc.spbbatchitemprocessor;

import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import javax.sql.DataSource;

import org.springframework.batch.core.Job;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing;
import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepScope;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.item.database.JdbcPagingItemReader;
import org.springframework.batch.item.database.Order;
import org.springframework.batch.item.database.support.MySqlPagingQueryProvider;
import org.springframework.batch.item.file.FlatFileItemWriter;
import org.springframework.batch.item.file.transform.LineAggregator;
import org.springframework.batch.item.support.ClassifierCompositeItemWriter;
import org.springframework.batch.item.support.CompositeItemProcessor;
import org.springframework.batch.item.support.CompositeItemWriter;
import org.springframework.batch.item.xml.StaxEventItemWriter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.classify.Classifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.io.FileSystemResource;
import org.springframework.jdbc.core.RowMapper;
import org.springframework.oxm.xstream.XStreamMarshaller;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;

@Configuration
@EnableBatchProcessing
public class ItemProcessorDemo {

	@Autowired
	private JobBuilderFactory jobBuilderFactory;

	@Autowired
	private StepBuilderFactory stepBuilderFactory;

	@Autowired
	private ItemProcessor<Customer, Customer> firstNameUpperProcessor;

	@Autowired
	private ItemProcessor<Customer, Customer> idFilterProcessor;

	@Autowired
	private DataSource dataSource;
	
	@Autowired
	@Qualifier("consoleItemWriter")
	private ConsoleItemWriter consoleItemWriter;

	@Autowired
	@Qualifier("multiFileItemWriter")
	private ItemWriter<? super Customer> multiFileItemWriter;

	@Bean
	public Job itemProcessorDemoJob() {
		return jobBuilderFactory.get("itemProcessorDemoJob")
				.start(itemProcessorDemoStep())
				.build();
	}

	@Bean
	public Step itemProcessorDemoStep() {
		return stepBuilderFactory.get("iItemProcessorDemoStep")
				.<Customer, Customer>chunk(5).
				reader(dbJdbcReader())
				// .processor(firstNameUpperProcessor)
				.processor(getprocessors())
				.writer(multiFileItemWriter)
				.build();
	}

	@Bean
	public CompositeItemProcessor<Customer, Customer> getprocessors() {
		CompositeItemProcessor<Customer, Customer> processor = new CompositeItemProcessor<Customer, Customer>();
		List<ItemProcessor<Customer, Customer>> delegates = new ArrayList<>();
		delegates.add(firstNameUpperProcessor);
		delegates.add(idFilterProcessor);
		processor.setDelegates(delegates);
		return processor;
	}

	@Bean
	@StepScope
	public JdbcPagingItemReader<Customer> dbJdbcReader() {
		JdbcPagingItemReader<Customer> reader = new JdbcPagingItemReader<Customer>();
		reader.setDataSource(dataSource);
		// 设置读取缓存
		reader.setFetchSize(2);
		// 把读取到的记录转换成User对象
		reader.setRowMapper(new RowMapper<Customer>() {

			@Override
			public Customer mapRow(ResultSet resultSet, int rowNum) throws SQLException {
				Customer customer = new Customer();
				customer.setId(resultSet.getLong("id"));
				customer.setFirstName(resultSet.getString("firstName"));
				customer.setLastName(resultSet.getString("lastName"));
				customer.setBirthday(resultSet.getString("birthday"));
				return customer;
			}
		});
		// 指定sq1语句
		MySqlPagingQueryProvider provider = new MySqlPagingQueryProvider();
		provider.setSelectClause("id,firstName,lastName,birthday");
		provider.setFromClause("from customer");
		// 指定根据哪个字段进行排序
		Map<String, Order> sort = new HashMap<>(1);
		sort.put("id", Order.ASCENDING);
		provider.setSortKeys(sort);
		reader.setQueryProvider(provider);
		return reader;
	}

	// fileWriter 输出数据到txt文件
	@Bean
	public FlatFileItemWriter<Customer> jsonFileWriter() {
		// 把Customer对象转成字符串输出到文件
		FlatFileItemWriter<Customer> writer = new FlatFileItemWriter<Customer>();
		String path = "E:\\test/customer.txt";
		writer.setResource(new FileSystemResource(path));
		// 把Customer对象转成字符串
		writer.setLineAggregator(new LineAggregator<Customer>() {
			ObjectMapper mapper = new ObjectMapper();

			@Override
			public String aggregate(Customer customer) {
				String str = null;
				try {
					str = mapper.writeValueAsString(customer);
				} catch (JsonProcessingException e) {
					e.printStackTrace();
				}

				return str;
			}
		});
		try {
			writer.afterPropertiesSet();
		} catch (Exception e) {
			e.printStackTrace();
		}
		return writer;
	}

	@Bean
	public StaxEventItemWriter<Customer> xmlWriter() {
		StaxEventItemWriter writer = new StaxEventItemWriter<Customer>();
		XStreamMarshaller marshaller = new XStreamMarshaller();
		// 告诉marshaller把数据转成什么类型
		Map<String, Class> map = new HashMap<>();
		map.put("customer", Customer.class);
		marshaller.setAliases(map);

		writer.setRootTagName("customers");
		writer.setMarshaller(marshaller);

		String path = "E:\\test/cus.xml";
		writer.setResource(new FileSystemResource(path));
		try {
			writer.afterPropertiesSet();
		} catch (Exception e) {
			e.printStackTrace();
		}
		return writer;
	}

	//组合输出composite
	@Bean
	public CompositeItemWriter<Customer> multiFileItemWriter() {
		CompositeItemWriter<Customer> writer = new CompositeItemWriter<Customer>();
		// 输出到两个不同的文件中
		writer.setDelegates(Arrays.asList(jsonFileWriter(), xmlWriter()));
		try {
			writer.afterPropertiesSet();
		} catch (Exception e) {
			e.printStackTrace();
		}
		return writer;
	}
	
	//过滤式组合composite
	@Bean
    public ClassifierCompositeItemWriter<Customer> multiFileItemWriter2(){
        ClassifierCompositeItemWriter<Customer> writer = new ClassifierCompositeItemWriter<Customer>();
        writer.setClassifier(new Classifier<Customer, ItemWriter<? super Customer>>() {
            @Override
            /**
             * @Description: classify 分类，比如按照年龄分成两个文件
             * @Param: [customer]
             * @Return: org.springframework.batch.item.ItemWriter<? super com.example.springbatchdemo.pojo.Customer>
             */
            public ItemWriter<? super Customer> classify(Customer customer) {
                //按照Customer的id进行分类
                ItemWriter<Customer> writer1 = customer.getId()%2==0?jsonFileWriter():xmlWriter();
                return writer1;
            }
        });
        return writer;
    }
}
